package com.carryking.smart.rpc.discover.impl;

import com.carryking.smart.rpc.discover.DiscoverService;
import com.carryking.smart.rpc.entity.DiscoverConfigEntity;
import com.carryking.smart.rpc.entity.ServiceInfoEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基于zookeeper的服务发现
 * Created by jincarry on 2018/1/11.
 */
@Slf4j
public class ZkDiscoverService implements DiscoverService {

    private static final String SERVICE_NAMESPACE = "smart-rpc";

    private static final String SERVICE_BASIC_PATH = "/service/center/";

    private CuratorFramework client;

    private static Map<String, List<String>> serviceRepository = new ConcurrentHashMap<String, List<String>>();

    private String zkConnectString;

    private String localIp;

    private Integer serverPort;

    public ZkDiscoverService(DiscoverConfigEntity configEntity) {
        this.zkConnectString = configEntity.getHost();
        try {
            InetAddress addr = InetAddress.getLocalHost();
            localIp = addr.getHostAddress();
            this.serverPort = configEntity.getPort();
        } catch (UnknownHostException e) {
            log.error("初始化服务注册中心时失败,获取本机IP时异常");
            throw new RuntimeException("初始化服务注册中心时失败,获取本机IP时异常");
        }

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        final CuratorFramework initClient = CuratorFrameworkFactory.builder().sessionTimeoutMs(1000).connectString(zkConnectString).retryPolicy(retryPolicy).namespace(SERVICE_NAMESPACE).build();
        initClient.start();
        this.client = initClient;
        try {

            Stat stat = client.checkExists().forPath(StringUtils.left(SERVICE_BASIC_PATH,SERVICE_BASIC_PATH.length() -1));
            if (stat == null) {
                client.create().creatingParentContainersIfNeeded().forPath(StringUtils.left(SERVICE_BASIC_PATH,SERVICE_BASIC_PATH.length() -1));
            }

            List<String> serviceList = client.getChildren().forPath(StringUtils.left(SERVICE_BASIC_PATH,SERVICE_BASIC_PATH.length() -1));
            for (String item : serviceList) {
                addWatch(client, item);
                List<String> serviceItemList = client.getChildren().forPath(SERVICE_BASIC_PATH + item);
                if (CollectionUtils.isEmpty(serviceItemList)) {
                    continue;
                }
                List<String> strings = new ArrayList<String>(serviceItemList.size());
                for (String itemDetail : serviceItemList) {
                    byte[] data = client.getData().forPath(SERVICE_BASIC_PATH + item + "/" + itemDetail);
                    if (data != null && data.length > 0) {
                        strings.add(new String(data));
                    }
                }
                serviceRepository.put(item, strings);
            }
        } catch (Exception e) {
            log.error("服务注册中心初始化失败",e);
        }
    }

    public boolean regisService(String serviceName) {
        if (StringUtils.isBlank(serviceName)) {
            log.warn("服务注册失败,服务名不能为空");
            return false;
        }

        try {
            Stat stat = client.checkExists().forPath(SERVICE_BASIC_PATH + serviceName);
            if (stat == null) {
                client.create().creatingParentContainersIfNeeded().forPath(SERVICE_BASIC_PATH + serviceName,serviceName.getBytes());
            }
            client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(SERVICE_BASIC_PATH + serviceName + "/" + serviceName, getClientInfo().getBytes("UTF-8"));
            if (log.isDebugEnabled()) {
                log.debug("服务:{}注册成功-->{}:{}",serviceName,localIp,serverPort);
            }
            return true;
        } catch (Exception e) {
            log.error("服务注册时发生异常;serviceName={}",serviceName,e);
            client.close();
        }
        return false;
    }

    public ServiceInfoEntity getServiceInfoByName(String serviceName) {

        try {
            Stat stat = client.checkExists().forPath(SERVICE_BASIC_PATH + serviceName);
            if (stat == null) {
                return null;
            }
            List<String> serviceList = serviceRepository.get(serviceName);
            if (CollectionUtils.isEmpty(serviceList)) {
                log.warn("没有在注册中心发现服务:serviceName={}", serviceName);
                return null;
            }
            //简单方式处理负载
            int index = RandomUtils.nextInt(0, serviceList.size());
            String serviceAddressStr = serviceList.get(index);
            String[] data = serviceAddressStr.split(":");
            if (data.length != 2) {
                log.warn("解析服务提供者地址时异常,原始字符串为:{}",serviceAddressStr);
                return null;
            }
            ServiceInfoEntity serviceAddress = new ServiceInfoEntity();
            serviceAddress.setIp(data[0]);
            serviceAddress.setPort(Integer.parseInt(data[1]));
            return serviceAddress;
        } catch (Exception e) {
            log.warn("获取服务地址时异常serviceName={}",serviceName,e);
        }
        return null;
    }


    public void addWatch(final CuratorFramework client, final String serviceName) throws Exception {
        client.getChildren().usingWatcher(new Watcher() {
            public void process(WatchedEvent event) {
                if (log.isDebugEnabled()) {
                    log.debug("服务列表发生变化,serviceName={}",serviceName);
                }
                try {
                    List<String> strings = client.getChildren().forPath(SERVICE_BASIC_PATH + serviceName );
                    List<String> datas = new ArrayList<String>(strings.size());
                    for (String item : strings) {
                        byte[] data = client.getData().forPath(SERVICE_BASIC_PATH + serviceName + "/" + item);
                        datas.add(new String(data));
                    }
                    serviceRepository.remove(serviceName);
                    serviceRepository.put(serviceName, datas);
                    if (log.isDebugEnabled()) {
                        log.debug("serviceName={},serviceList={}",serviceName,datas);
                    }
                    addWatch(client,serviceName);
                } catch (Exception e) {
                    log.warn("zookeeper监听服务改变时发生异常", serviceName, e);
                    client.close();
                }
            }
        }).forPath(SERVICE_BASIC_PATH + serviceName);
    }

    private String getClientInfo() {
        return String.format("%s:%s", localIp, serverPort);
    }
}
